package dht.server;

import java.util.Vector;

import dht.AbstractTransactionHandler;
import dht.IDataServer;
import dht.IDataServerRouteTable;
import dht.IStorage;
import dht.entity.DataObject;
import dht.entity.GlobalVariables;
import dht.entity.Message;
import dht.entity.MessageType;
import dht.entity.NodeInfo;
import dht.socket.SocketHandler;
import dht.socket.TcpConnection;
import dht.throwable.DhtException;

public class DataServerTransactionHandlerImpl extends
		AbstractTransactionHandler {

	protected IStorage storage;
	protected IDataServerRouteTable routeTable;
	protected IDataServer dataServer;

	public DataServerTransactionHandlerImpl(NodeInfo localInfo,
			NodeInfo remoteInfo, IDataServerRouteTable routeTable,
			IDataServer dataServer, TcpConnection tcpConnection,
			IStorage storage) {
		super(localInfo, remoteInfo, tcpConnection);
		this.routeTable = routeTable;
		this.dataServer = dataServer;
		this.storage = storage;
	}

	public DataServerTransactionHandlerImpl(NodeInfo localInfo,
			SocketHandler socketHandler, IDataServerRouteTable routeTable,
			IDataServer dataServer, TcpConnection tcpConnection,
			IStorage storage) {
		super(localInfo, socketHandler, tcpConnection);
		this.routeTable = routeTable;
		this.dataServer = dataServer;
		this.storage = storage;
	}

	@Override
	protected void runInitiative(Message msg) {
		switch (msg.getMsgType()) {
		case SERVER_TABLE_REQ:
			doServerTableReq(msg);
			break;
		case JOIN_REQ:
			doJoinReq(msg);
			break;
		default:
			break;
		}
	}

	@Override
	protected void runPassive() {
		Message reqMsg = null;
		try {
			reqMsg = socketHandler.readMessage();
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		switch (reqMsg.getMsgType()) {
		case READ_REQ:
			handleReadReq(reqMsg);
			break;
		case WRITE_REQ:
			handleWriteReq(reqMsg);
			break;
		case TABLE_PUSH_REQ:
			handleTablePushReq(reqMsg);
			break;
		case MOV_REQ:
			handleMovReq(reqMsg);
			break;
		case DATA_MOV_REQ:
			handleDataMovReq(reqMsg);
			break;
		default:
			break;
		}
	}

	protected void doJoinReq(Message msg) {
		SocketHandler socketHandler = null;
		try {
			socketHandler = tcpConnection
					.genTransactionSocketHandler(remoteInfo);
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		try {
			socketHandler.sendMessage(msg);
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		Message respMsg = null;
		try {
			respMsg = socketHandler.readMessage();
			assert (respMsg.getMsgType() == MessageType.JOIN_RESP);
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		dataServer.setRouteTable(respMsg.getDataServerRouteTable());
		Message retMsg = new Message(msg.getTransactionId(),
				MessageType.JOIN_ACK, localInfo);
		try {
			socketHandler.sendMessage(retMsg);
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		try {
			socketHandler.closeSocket();
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
	}

	protected void doServerTableReq(Message msg) {
		SocketHandler socketHandler = null;
		try {
			socketHandler = tcpConnection
					.genTransactionSocketHandler(remoteInfo);
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		try {
			socketHandler.sendMessage(msg);
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		Message respMsg = null;
		try {
			respMsg = socketHandler.readMessage();
			assert (respMsg.getMsgType() == MessageType.SERVER_TABLE_RESP);
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		dataServer.setRouteTable(respMsg.getDataServerRouteTable());
		try {
			socketHandler.closeSocket();
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
	}

	protected void handleTablePushReq(Message msg) {
		dataServer.setRouteTable(msg.getDataServerRouteTable());
		Message respMsg = new Message(msg.getTransactionId(),
				MessageType.TABLE_PUSH_RESP, localInfo);
		try {
			socketHandler.sendMessage(respMsg);
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		try {
			socketHandler.closeSocket();
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
	}

	protected void handleDataMovReq(Message msg) {
		int cnt = 0;
		int total = -1;
		do {
			for (DataObject dataObject : msg.getDataObjects()) {
				try {
					storage.write(dataObject, false);
				} catch (DhtException e) {
					e.printStackTrace();
					return;
				}
			}
			Message retMsg = new Message(msg.getTransactionId(),
					MessageType.DATA_MOV_RESP, localInfo);
			try {
				socketHandler.sendMessage(retMsg);
			} catch (DhtException e) {
				e.printStackTrace();
				return;
			}
			cnt++;
			if (msg.getDataObjects() == null
					|| msg.getDataObjects().size() == 0) {
				total = msg.getMoveId();
			}
			if (total == cnt) {
				break;
			}
			try {
				msg = socketHandler.readMessage();
				// System.out.println("msg size: " + msg.getMessageSize());
				assert (msg.getMsgType() == MessageType.DATA_MOV_REQ);
			} catch (DhtException e) {
				e.printStackTrace();
				return;
			}
		} while (true);

		try {
			socketHandler.closeSocket();
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
	}

	protected void handleMovReq(Message msg) {
		Vector<String> fileNames = null;
		try {
			fileNames = storage.getFileNamesByVnodeIds(msg
					.getMovementVnodeIds());
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		SocketHandler nextSocketHandler = null;
		try {
			nextSocketHandler = tcpConnection.genTransactionSocketHandler(msg
					.getMoveToServerInfo());
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		int moveId = 1;
		int msgPending = 0;
		for (int i = 0; i < fileNames.size();) {
			Message newMsg = new Message(msg.getTransactionId(),
					MessageType.DATA_MOV_REQ, localInfo);
			Vector<DataObject> dataObjects = new Vector<DataObject>();
			dataObjects.clear();
			int totalNum = 0;
			do {
				String fileName = fileNames.get(i);
				DataObject dataObject = null;
				try {
					dataObject = storage.read(fileName);
				} catch (DhtException e) {
					e.printStackTrace();
					return;
				}
				dataObjects.add(dataObject);
				++i;
				++totalNum;
			} while (i < fileNames.size()
					&& totalNum < GlobalVariables.moveNumber);
			newMsg.setDataObjects(dataObjects);
			newMsg.setMoveId(moveId++);
			try {
				nextSocketHandler.sendMessage(newMsg);
				msgPending++;
				if (msgPending > GlobalVariables.moveConcurrentNumber) {
					Message respMsg = nextSocketHandler.readMessage();
					assert (respMsg.getMsgType() == MessageType.DATA_MOV_RESP);
					msgPending--;
				}
			} catch (DhtException e) {
				e.printStackTrace();
				return;
			}
		}
		Message newMsg = new Message(msg.getTransactionId(),
				MessageType.DATA_MOV_REQ, localInfo);
		Vector<DataObject> dataObjects = new Vector<DataObject>();
		dataObjects.clear();
		newMsg.setDataObjects(dataObjects);
		newMsg.setMoveId(moveId++);
		try {
			nextSocketHandler.sendMessage(newMsg);
			msgPending++;
			while (msgPending > 0) {
				Message respMsg = nextSocketHandler.readMessage();
				assert (respMsg.getMsgType() == MessageType.DATA_MOV_RESP);
				msgPending--;
			}
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		try {
			nextSocketHandler.closeSocket();
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		Message retMsg = new Message(msg.getTransactionId(),
				MessageType.MOV_RESP, localInfo);
		Message respMsg = null;
		try {
			socketHandler.sendMessage(retMsg);
			respMsg = socketHandler.readMessage();
			assert (respMsg.getMsgType() == MessageType.DATA_RELEASE_REQ);
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		dataServer.setRouteTable(respMsg.getDataServerRouteTable());
		// TODO real data release
		retMsg = new Message(msg.getTransactionId(),
				MessageType.DATA_RELEASE_RESP, localInfo);
		try {
			socketHandler.sendMessage(retMsg);
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		try {
			socketHandler.closeSocket();
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
	}

	protected void handleReadReq(Message msg) {
		Message respMsg = null;
		if (routeTable.isInLocal(msg.getFileName(), localInfo)) {
			respMsg = new Message(msg.getTransactionId(),
					MessageType.READ_RESP, localInfo);
			respMsg.setFileName(msg.getFileName());
			try {
				respMsg.setDataObject(storage.read(msg.getFileName()));
			} catch (DhtException e) {
				respMsg.setDataObject(null);
				// file not exist
				// e.printStackTrace();
			}
		} else {
			respMsg = new Message(msg.getTransactionId(), MessageType.READ_FWD,
					localInfo);
			respMsg.setFileName(msg.getFileName());
			NodeInfo fwdServerInfo = routeTable
					.getServerList(msg.getFileName()).get(0);
			respMsg.setFwdServerInfo(fwdServerInfo);
		}
		try {
			socketHandler.sendMessage(respMsg);
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		try {
			socketHandler.closeSocket();
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
	}

	protected void handleWriteReq(Message msg) {
		Message respMsg = null;
		if (routeTable.isInLocal(msg.getFileName(), localInfo)) {
			respMsg = new Message(msg.getTransactionId(),
					MessageType.WRITE_RESP, localInfo);
			respMsg.setFileName(msg.getFileName());
			try {
				storage.write(msg.getDataObject(), false);
			} catch (DhtException e) {
				e.printStackTrace();
				return;
			}
		} else {
			respMsg = new Message(msg.getTransactionId(),
					MessageType.WRITE_FWD, localInfo);
			respMsg.setFileName(msg.getFileName());
			NodeInfo fwdServerInfo = routeTable
					.getServerList(msg.getFileName()).get(0);
			respMsg.setFwdServerInfo(fwdServerInfo);
		}
		try {
			socketHandler.sendMessage(respMsg);
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
		try {
			socketHandler.closeSocket();
		} catch (DhtException e) {
			e.printStackTrace();
			return;
		}
	}

}
